package icu.zlz.emqx.mqtt;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import icu.zlz.common.core.constant.MqttConstants;
import icu.zlz.emqx.domain.IotDevices;
import icu.zlz.emqx.domain.IotGroup;
import icu.zlz.emqx.mapper.IotDevicesMapper;
import icu.zlz.emqx.mapper.IotGroupMapper;
import icu.zlz.emqx.service.impl.EmqxService;
import org.apache.commons.lang3.ObjectUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

/**
 * mqtt
 *
 * @author zangsheng
 */
@Component
public class MQTTMsg {

    /**
     * $SYS/brokers/{node}/clients/{clientid}/connected 系统主题，上线事件。当任意客户端上线时，EMQX 就会发布该主题的消息
     * ^\\$SYS/brokers/([^/]+)/clients/([^/]+)/connected$
     * ^ 表示字符串的开始。
     * \\$SYS/brokers/ 匹配固定的 $SYS/brokers/ 部分。
     * ([^/]+) 匹配 {node} 部分，[^/]+ 表示一个或多个非 / 的字符。
     * /clients/ 匹配固定的 /clients/ 部分。
     * ([^/]+) 匹配 {clientid} 部分。
     * /connected$ 匹配固定的 /connected 部分，$ 表示字符串的结束。
     */
    private static final String TOPIC_PATTERN_CONNECTED = "^\\$SYS/brokers/([^/]+)/clients/([^/]+)/connected$";
    //离线事件
    private static final String TOPIC_PATTERN_DISCONNECTED = "^\\$SYS/brokers/([^/]+)/clients/([^/]+)/disconnected$";


    @Resource
    private IotGroupMapper iotGroupMapper;

    @Resource
    private IotDevicesMapper iotDevicesMapper;
    @Resource
    private EmqxService emqxService;

    @Resource
    private MqttClient mqttClient;

    @PostConstruct
    public void init() {

        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {

            }

            /**
             * 处理mqtt信息
             * @param topic 主题
             * @param mqttMessage 信息
             * @throws Exception
             */
            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                String msg = new String(mqttMessage.getPayload(), MqttConstants.MSG_CODE);
                Map<String, Object> map = JSON.parseObject(msg, new TypeReference<Map<String, Object>>() {
                });

                //用户上下线
                if (Pattern.compile(TOPIC_PATTERN_CONNECTED).matcher(topic).matches() || Pattern.compile(TOPIC_PATTERN_DISCONNECTED).matcher(topic).matches()) {
                        onlineAndOffline(map);


                }
                if (Pattern.compile("^" + MqttConstants.MQTT_ALARM_TOPIC + "/([^/]+)/([^/]+)/([^/]+)$").matcher(topic).matches()) {
                    //告警
                    alarm(map);
                }

            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        });
    }

    private void alarm(Map<String, Object> map) {
        //发送告警通知
    }

    private void onlineAndOffline(Map<String, Object> message) {
        String username = (String) message.get("username");
        String clientid = (String) message.get("clientid");
        Long connected_at = (Long) message.get("connected_at");
        Long disconnected_at = (Long) message.get("disconnected_at");
        IotGroup queryGroup = new IotGroup();
        queryGroup.setSecretKey(username);
        List<IotGroup> iotGroups = iotGroupMapper.selectIotGroupList(queryGroup);
        IotDevices queryDevice = new IotDevices();
        queryDevice.setDeviceClientId(clientid);
        List<IotDevices> iotDevices = iotDevicesMapper.selectIotDevicesList(queryDevice);
        String subscribeTopic;
        if (!CollectionUtils.isEmpty(iotGroups) && !CollectionUtils.isEmpty(iotDevices)) {
            IotDevices iotDevice = iotDevices.get(0);
            if (ObjectUtils.isNotEmpty(connected_at)) {
                //上线
                iotDevice.setLastConnectionTime(new Date(connected_at));
                subscribeTopic = MqttConstants.MQTT_CLIENT_TOPIC + "/" + iotDevice.getUid() + "/" + clientid;
                emqxService.clientSubscribe(clientid, subscribeTopic, 2, null, null, null);
            } else {
                //下线
                iotDevice.setLastDisconnectionTime(new Date(disconnected_at));
            }
            //更新时间
            iotDevicesMapper.updateIotDevices(iotDevice);
        }
    }
}
